@ConcurrencyPractitioner please also take a look

Sure, I have no problem with this. Thanks for picking this up. :)

rerun cpp tests

@srkukarni @sijie can you look at this again? Thanks

Alternatively, instead of allowing user's to be able to have Message as input to a function, a simpler approach of we would be just to add this API to Context:

 Message getInternalMessage();

and the Impl would be:

Message getInternalMessage() {
return ((PulsarRecord) this.record).getMessage);
}

There also isn't any good way (without breaking BC) for the WindowFunction API to support Message as a input since the WindowFunction API is

public interface WindowFunction<I, O> {
    /**
     * Process the input.
     *
     * @return the output
     */
    O process(Collection<Record<I>> input, WindowContext context) throws Exception;
}

I don't think it makes much sense to do

    O process(Collection<Record<Message<I>>> input, WindowContext context) throws Exception;

We would have to change the interface (BC breaking) to:

public interface WindowFunction<I, O> {
    /**
     * Process the input.
     *
     * @return the output
     */
    O process(Collection<I> input, WindowContext context) throws Exception;
}

To support Message as a input

Alternatively, we can change the API for WindowFunction to be something like

    O process(Window<I> input, WindowContext context) throws Exception;

and you can do

input.getAsMessages()

or 

input.getAsRecords()

or in the WindowContext we implement a method like

context.getWindowAsMessages()

The simplest and non-BC-breaking approach I can think of for Functions and WindowFunctions:

just to add these getter methods to Context and WindowContext:

Message getInternalMessage();

and

List<Message> getWindowAsMessages()

@jerrypeng What about the message type? I believe Context is not generic at this point. Would you get a Message<?> then?

@merlimat that is correct which is a con for that approach.

@sijie @srkukarni @merlimat what we can do is this:
#4341

simpler approach that solves all our problems

+1 for #4341